--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
Commit 1f089cd3fd8d90468a863a2301e374e4e48a2735
Parents : 14b1b78
Author : Ivan <ivan@quad4.io>
Signature : Invalid signer <e46112d44649266d71fe2193e00a4710>, author is <ivan@quad4.io>
Date : 2026-06-22T17:49:47-05:00
refactor(tests): streamline fuzzing tests by removing obsolete cases
Changes
Diff
diff --git a/tests/backend/test_fuzzing.py b/tests/backend/test_fuzzing.py
index c41f60a2..e1e9d567 100644
--- a/tests/backend/test_fuzzing.py
+++ b/tests/backend/test_fuzzing.py
@@ -2,11 +2,9 @@
import os
import random
-import time
from contextlib import ExitStack
from unittest.mock import MagicMock, patch
-import LXMF
import pytest
import RNS
from hypothesis import HealthCheck, given, settings
@@ -23,12 +21,7 @@ from meshchatx.src.backend.meshchat_utils import (
parse_lxmf_display_name,
parse_nomadnetwork_node_display_name,
)
-from meshchatx.src.backend.nomadnet_utils import (
- convert_nomadnet_field_data_to_map,
- convert_nomadnet_string_data_to_map,
-)
from meshchatx.src.backend.page_node import normalize_page_filename
-from meshchatx.src.backend.telemetry_utils import Telemeter
@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
@@ -41,54 +34,15 @@ def test_interface_config_parsing_fuzzing(config_text):
pytest.fail(f"InterfaceConfigParser crashed with input: {e}")
-@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
-@given(identity_bytes=st.binary(min_size=0, max_size=2048))
-def test_identity_parsing_fuzzing(identity_bytes):
- """Fuzz RNS.Identity loading with random bytes."""
- try:
- RNS.Identity.from_bytes(identity_bytes)
- except Exception:
- # RNS.Identity.from_bytes is expected to fail on random bytes,
- # but it should not cause an unhandled crash of the process.
- pass
-
-
-@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
-@given(path_data=st.one_of(st.none(), st.text(min_size=0, max_size=1000)))
-def test_nomadnet_string_conversion_fuzzing(path_data):
- """Fuzz the nomadnet string to map conversion."""
- try:
- convert_nomadnet_string_data_to_map(path_data)
- except Exception as e:
- pytest.fail(
- f"convert_nomadnet_string_data_to_map crashed with data {path_data}: {e}",
- )
-
-
-@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
-@given(
- field_data=st.one_of(
- st.none(),
- st.dictionaries(keys=st.text(), values=st.text()),
- st.text(),
- ),
-)
-def test_nomadnet_field_conversion_fuzzing(field_data):
- """Fuzz the nomadnet field data to map conversion."""
- try:
- convert_nomadnet_field_data_to_map(field_data)
- except Exception as e:
- pytest.fail(
- f"convert_nomadnet_field_data_to_map crashed with data {field_data}: {e}",
- )
-
-
@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
@given(name=st.text(min_size=0, max_size=500))
def test_normalize_page_filename_fuzzing(name):
"""Fuzz mesh server page filename normalization."""
try:
- normalize_page_filename(name)
+ result = normalize_page_filename(name)
+ assert isinstance(result, str)
+ assert "/" not in result
+ assert "\\" not in result
except ValueError:
pass
@@ -104,38 +58,6 @@ def test_display_name_parsing_fuzzing(app_data_base64):
pytest.fail(f"Display name parsing crashed with data {app_data_base64}: {e}")
-@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
-@given(
- fields_data=st.dictionaries(
- st.integers(min_value=0, max_value=255),
- st.binary(min_size=0, max_size=1000),
- ),
-)
-def test_lxmf_fields_parsing_fuzzing(fields_data):
- """Fuzz the parsing of LXMF message fields."""
- try:
- for field_id, field_data in fields_data.items():
- if field_id == 0x01: # FIELD_COMMANDS
- try:
- import umsgpack
-
- commands = umsgpack.unpackb(field_data)
- if isinstance(commands, list):
- for cmd in commands:
- if isinstance(cmd, dict):
- for k, v in cmd.items():
- pass
- elif isinstance(commands, dict):
- for k, v in commands.items():
- pass
- except Exception:
- pass
- elif field_id == 0x02: # FIELD_TELEMETRY
- Telemeter.from_packed(field_data)
- except Exception:
- pass
-
-
@pytest.fixture
def temp_dir(tmp_path):
return str(tmp_path)
@@ -143,7 +65,6 @@ def temp_dir(tmp_path):
@pytest.fixture
def mock_app(temp_dir):
- # Save real Identity class to use as base for our mock class
real_identity_class = RNS.Identity
class MockIdentityClass(real_identity_class):
@@ -152,7 +73,6 @@ def mock_app(temp_dir):
self.hexhash = self.hash.hex()
with ExitStack() as stack:
- # Mock database and other managers to avoid heavy initialization
stack.enter_context(patch("meshchatx.src.backend.identity_context.Database"))
stack.enter_context(
patch("meshchatx.src.backend.identity_context.ConfigManager"),
@@ -244,7 +164,6 @@ def mock_app(temp_dir):
patch.object(MockIdentityClass, "from_bytes", return_value=mock_id),
)
- # Make run_async a no-op that doesn't trigger coroutine warnings
def mock_run_async(coro):
import asyncio
@@ -259,7 +178,6 @@ def mock_app(temp_dir):
reticulum_config_dir=temp_dir,
)
- # Setup config mock to return real values to avoid JSON serialization issues
app.config = MagicMock()
app.config.display_name.get.return_value = "Test User"
app.config.auto_announce_enabled.get.return_value = True
@@ -327,7 +245,6 @@ def test_announce_overload(mock_app, num_announces):
aspect = "lxmf.delivery"
app_data = b"test_app_data"
- # Mock database to return a valid announce dict
mock_app.database.announces.get_announce_by_hash.return_value = {
"aspect": "lxmf.delivery",
"destination_hash": "some_hash",
@@ -348,7 +265,6 @@ def test_announce_overload(mock_app, num_announces):
announce_packet_hash,
)
- # Verify that the database was called for each announce
assert mock_app.announce_manager.upsert_announce.call_count == num_announces
@@ -364,7 +280,7 @@ def test_message_spamming(mock_app, num_messages):
mock_message = MagicMock()
mock_message.source_hash = os.urandom(16)
mock_message.hash = os.urandom(16)
- mock_message.get_fields.return_value = {} # No telemetry field
+ mock_message.get_fields.return_value = {}
mock_message.title = f"Spam Title {i}"
mock_message.content = f"Spam Content {i}"
@@ -395,171 +311,6 @@ def test_message_spamming_large_payloads(mock_app, num_messages, payload_size):
assert mock_app.db_upsert_lxmf_message.call_count == num_messages
-@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
-@given(
- msg=st.dictionaries(
- keys=st.text(),
- values=st.one_of(
- st.text(),
- st.integers(),
- st.booleans(),
- st.dictionaries(keys=st.text(), values=st.text()),
- ),
- min_size=1,
- max_size=10,
- ).flatmap(
- lambda d: st.sampled_from(
- [
- "ping",
- "config.set",
- "nomadnet.download.cancel",
- "nomadnet.page.archives.get",
- "lxmf.forwarding.rule.add",
- "lxmf.forwarding.rule.delete",
- "lxm.ingest_uri",
- "lxm.generate_paper_uri",
- "keyboard_shortcuts.get",
- "telephone.recordings.get",
- ],
- ).map(lambda t: {**d, "type": t}),
- ),
-)
-@pytest.mark.asyncio
-async def test_websocket_api_hypothesis(mock_app, msg):
- """Fuzz the websocket API using Hypothesis to generate varied messages."""
- # Use MagicMock instead of AsyncMock to avoid coroutine warnings
- mock_client = MagicMock()
- mock_client.send_str = MagicMock(side_effect=lambda data: None)
- try:
- await mock_app.on_websocket_data_received(mock_client, msg)
- except Exception:
- pass
-
-
-@pytest.mark.asyncio
-async def test_websocket_api_fuzzing(mock_app):
- """Fuzz the websocket API with various message types and payloads."""
- # Use MagicMock instead of AsyncMock to avoid coroutine warnings
- mock_client = MagicMock()
- mock_client.send_str = MagicMock(side_effect=lambda data: None)
-
- # Test cases with different message types and malformed/unexpected data
- fuzz_messages = [
- {"type": "ping"},
- {"type": "config.set", "config": {"invalid_key": "invalid_value"}},
- {"type": "config.set", "config": "not_a_dict"},
- {"type": "nomadnet.download.cancel", "download_id": "non_existent_id"},
- {
- "type": "nomadnet.page.archives.get",
- "destination_hash": "invalid_hash",
- "page_path": "/invalid",
- },
- {"type": "lxmf.forwarding.rule.add", "rule": {}},
- {"type": "lxmf.forwarding.rule.delete", "id": -1},
- {"type": "lxm.ingest_uri", "uri": "invalid_uri"},
- {
- "type": "lxm.generate_paper_uri",
- "destination_hash": "00" * 16,
- "content": "test",
- },
- {"type": "non_existent_type", "data": "random_data"},
- ]
-
- for msg in fuzz_messages:
- try:
- await mock_app.on_websocket_data_received(mock_client, msg)
- except Exception:
- pass
-
-
-@pytest.mark.asyncio
-async def test_config_fuzzing(mock_app):
- """Fuzz the config update logic with various values."""
- fuzz_configs = [
- {"display_name": "A" * 1000},
- {"display_name": None},
- {"auto_resend_failed_messages_when_announce_received": "not_a_bool"},
- {"unknown_config_option": 123},
- {},
- ]
-
- for config in fuzz_configs:
- try:
- if hasattr(mock_app, "update_config"):
- await mock_app.update_config(config)
- except Exception:
- pass
-
-
-def test_malformed_announce_data(mock_app):
- """Test handling of malformed or unexpected data in announces."""
- aspect = "lxmf.delivery"
- destination_hash = b"too_short" # Malformed hash
-
- # Test with None identity
- mock_app.on_lxmf_announce_received(aspect, destination_hash, None, None, b"")
-
- # Test with identity having None hash
- announced_identity = MagicMock()
- announced_identity.hash = None
- mock_app.on_lxmf_announce_received(
- aspect,
- destination_hash,
- announced_identity,
- None,
- b"",
- )
-
-
-def test_malformed_message_data(mock_app):
- """Test handling of malformed LXMF messages."""
- mock_message = MagicMock()
- del mock_message.source_hash
- mock_app.on_lxmf_delivery(mock_message)
-
-
-@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
-@given(
- weird_string=st.text(min_size=0, max_size=1000),
- large_binary=st.binary(min_size=0, max_size=10000),
-)
-def test_database_dao_fuzzing(mock_app, weird_string, large_binary):
- """Fuzz the database DAOs with weird strings and large binary data."""
- # Test AnnounceDAO
- announce_data = {
- "destination_hash": os.urandom(16).hex(),
- "aspect": weird_string,
- "identity_hash": weird_string,
- "identity_public_key": large_binary.hex(),
- "app_data": large_binary,
- "rssi": random.randint(-120, 0),
- "snr": random.uniform(-20, 20),
- "quality": random.uniform(0, 100),
- }
- try:
- mock_app.database.announces.upsert_announce(announce_data)
- except Exception:
- pass
-
- # Test MessageDAO
- message_data = {
- "hash": os.urandom(16).hex(),
- "source_hash": os.urandom(16).hex(),
- "destination_hash": os.urandom(16).hex(),
- "state": weird_string,
- "title": weird_string,
- "content": weird_string,
- "fields": {"weird": weird_string},
- "timestamp": time.time(),
- "is_incoming": random.choice([0, 1]),
- "is_spam": random.choice([0, 1]),
- }
- try:
- mock_app.database.messages.upsert_lxmf_message(message_data)
- except Exception:
- pass
-
-
@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
@given(
audio_bytes=st.binary(min_size=0, max_size=5000),
@@ -576,78 +327,3 @@ def test_lxmf_field_fuzzing(audio_bytes, image_bytes):
LxmfFileAttachment(file_name="test.txt", file_bytes=audio_bytes)
except Exception as e:
pytest.fail(f"LXMF field classes crashed: {e}")
-
-
-@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
-@given(command_bytes=st.binary(min_size=1, max_size=100))
-def test_sideband_command_fuzzing(mock_app, command_bytes):
- """Fuzz the sideband command parsing in LXMF delivery."""
- mock_message = MagicMock()
- mock_message.source_hash = os.urandom(16)
- mock_message.hash = os.urandom(16)
- # 0x01 is SidebandCommands.TELEMETRY_REQUEST
- mock_message.get_fields.return_value = {LXMF.FIELD_COMMANDS: [command_bytes]}
-
- try:
- mock_app.on_lxmf_delivery(mock_message)
- except Exception:
- pass
-
-
-@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
-@given(
- destination_hash=st.text(min_size=0, max_size=100),
- page_path=st.text(min_size=0, max_size=500),
- content=st.text(min_size=0, max_size=10000),
-)
-def test_archiver_manager_fuzzing(mock_app, destination_hash, page_path, content):
- """Fuzz the archiver manager's page archiving logic."""
- try:
- mock_app.archive_page(destination_hash, page_path, content, is_manual=True)
- except Exception:
- pass
-
-
-@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
-@given(state=st.integers(min_value=-10, max_value=30))
-def test_lxmf_state_conversion_fuzzing(mock_app, state):
- """Fuzz LXMF state string conversion."""
- mock_message = MagicMock()
- mock_message.state = state
- try:
- ReticulumMeshChat.convert_lxmf_state_to_string(mock_message)
- except Exception:
- pass
-
-
-@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
-@given(method=st.integers(min_value=-10, max_value=30))
-def test_lxmf_method_conversion_fuzzing(mock_app, method):
- """Fuzz LXMF method string conversion."""
- mock_message = MagicMock()
- mock_message.method = method
- try:
- ReticulumMeshChat.convert_lxmf_method_to_string(mock_message)
- except Exception:
- pass
-
-
-def test_telephone_announce_fuzzing(mock_app):
- """Fuzz telephone announce reception."""
- aspect = "telephone.call"
- destination_hash = os.urandom(16)
- announced_identity = MagicMock()
- announced_identity.hash = os.urandom(32)
- app_data = b"test_app_data"
- announce_packet_hash = os.urandom(16)
-
- try:
- mock_app.on_telephone_announce_received(
- aspect,
- destination_hash,
- announced_identity,
- app_data,
- announce_packet_hash,
- )
- except Exception:
- pass
──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────